#include <stdio.h>
#include <signal.h>
#include <time.h>

#include "st.h"
#include "ursax.h"
#include "debug.h"
#include "heartbeat.h"
#include "netserver.h"
#include "endpoints.h"
#include "log.h"
#include "ioengine.h"
#include "backend-fs.h"
#include "redis-client.h"
#include "incubator.h"
#include "concurrent-connection-pool.h"
#include "operations.h"
#include "statistics.h"
#include "chunkserver-journal/chunkserver-journal.h"

static bool append_mode = false;
static bool report_hdd = false;

struct NetServer *g_chunk_ns;

extern int g_disk_type;
extern uint32_t o_direct_size;

#define SERVER_UP_ENV_NAME "server_up"

static struct OperationTable operation_table[] = {
        { sizeof(struct Request_ChunkCreate),					&op_chunk_create},					//CHUNK_CREATE,
        { sizeof(struct Request_ChunkList),						&op_chunk_fake},					//CHUNK_LIST,
        { sizeof(struct Request_ChunkRead),						&op_chunk_fake},					//CHUNK_READ,
        { sizeof(struct Request_ChunkWrite),					&op_chunk_fake},					//CHUNK_WRITE,
        { sizeof(struct Request_ChunkWriteReplicate),			&op_chunk_fake},			//CHUNK_WRITE_REPLICATE,
        { sizeof(struct Request_ChunkCreateWriteReplicate),		&op_chunk_fake},	//CHUNK_CREATE_WRITE_REPLICATE,
        { sizeof(struct Request_ChunkGetVersion),				&op_chunk_get_version},				//CHUNK_GET_VERSION,
        { sizeof(struct Request_ChunkFlush),					&op_chunk_flush},					//CHUNK_FLUSH,
        { sizeof(struct Request_ChunkDelete),					&op_chunk_delete},					//CHUNK_DELETE,
        { sizeof(struct Request_ChunkChecksum),					&op_chunk_fake},				//CHUNK_CHECKSUM,
        { sizeof(struct Request_ChunkClone),					&op_chunk_fake},					//CHUNK_CLONE,
        { sizeof(struct Request_ChunkBreed),					&op_chunk_fake},					//CHUNK_BREED,
        { sizeof(struct Request_ChunkSuspendWrite),				&op_chunk_suspend_write},			//CHUNK_SUSPEND_WRITE,
        { sizeof(struct Request_NotifyNewReplica),				&op_chunk_fake},		//CHUNK_NOTIFY_NEW_REPLICA,
        { sizeof(struct Request_NotifyIncubationDone),			&op_chunk_fake},			//CHUNK_INCUBATION_DONE,
        { sizeof(struct Request_StartIncubation),				&op_chunk_fake},		//CHUNK_START_INCUBATION,
        { sizeof(struct Request_ChunkUndelete),					&op_chunk_fake},				//CHUNK_UNDELETE,
        { sizeof(struct Request_ChunkServerTerminate),			&op_chunk_fake},		//CHUNK_SERVER_TERMINATE,
        { sizeof(struct Request_ChunkServerPing),				&op_chunk_fake},				//CHUNK_SERVER_PING,
        { sizeof(struct Request_ChunkRename),					&op_chunk_fake},					//CHUNK_RENAME,
        { sizeof(struct Request_ChunkUpdate),					&op_chunk_fake},					//CHUNK_UPDATA
        { sizeof(struct Request_ChunkWriteReplicate3),			&op_chunk_fake},		// CHUNK_WRITE_REPLICATE3
        { sizeof(struct Request_ChunkWrite3),					&op_chunk_fake},					// CHUNK_WRITE3
        { sizeof(struct Request_ChunkRead3),					&op_chunk_fake},					// CHUNK_READ3
        { sizeof(struct Request_ChunkWriteReplicate2),			&op_chunk_fake},		// CHUNK_WRITE_REPLICATE2
        { sizeof(struct Request_ChunkSyncData),					&op_chunk_fake}, 				// CHUNK_SYNC_DATA
        { sizeof(struct Request_ChunkGetInfo),					&op_chunk_fake},				// CHUNK_GET_INFO
        { sizeof(struct Request_ChunkGetJournal),				&op_chunk_fake}, 			// CHUNK_GET_JOURNAL
        { sizeof(struct Request_ChunkServerRestart),			&op_chunk_fake},			// CHUNK_SERVER_RESTART
        { sizeof(struct Request_ChunkWriteReplicate4), 			&op_chunk_write_replicate4all},		// CHUNK_WRITE_REPLICATE4
        { sizeof(struct Request_ChunkWrite4), 					&op_chunk_write4},					// CHUNK_WRITE4
        { sizeof(struct Request_ChunkRead4), 					&op_chunk_read4},					// CHUNK_READ4
        { sizeof(struct Request_ChunkWriteECData), 				&op_chunk_fake},			// CHUNK_WRITE_EC_DATA
        { sizeof(struct Request_ChunkUpdateECParity), 			&op_chunk_fake},		// CHUNK_UPDATE_EC_PARITY
        { sizeof(struct Request_ChunkWriteECGroupData), 		&op_chunk_fake},		// CHUNK_WRITE_ECGROUP_DATA
        { sizeof(struct Request_ChunkWriteECOnErr), 			&op_chunk_fake},		// CHUNK_WRITE_EC_ON_ERROR
        { sizeof(struct Request_ChunkECFixCreate), 				&op_chunk_fake},				// CHUNK_FIX_CREATE
        { sizeof(struct Request_ChunkWriteECParity), 			&op_chunk_fake},			// CHUNK_WRITE_EC_PARITY
        { sizeof(struct Request_ChunkWriteEC), 					&op_chunk_fake},				// CHUNK_WRITE_EC
        { sizeof(struct Request_ChunkGetECVersion), 			&op_chunk_fake},			// CHUNK_GET_EC_VERSION
        { sizeof(struct Request_ChunkNotifyNewECChunk), 		&op_chunk_fake},			// CHUNK_NOTIFY_NEW_EC
        { sizeof(struct Request_ChunkCheckParityRight), 		&op_chunk_fake}, 		// CHECK_EC_PARITY_RIGHT
        { sizeof(struct Request_ChunkECFixCreate), 				&op_chunk_fake},				// CHUNK_FIX_CREATE2
        { sizeof(struct Request_ChunkCheckParityRight), 		&op_chunk_fake},		// CHECK_EC_PARITY_RIGHT2
        { sizeof(struct Request_ChunkClearEcRecord), 			&op_chunk_fake},			// CHUNK_CLEAR_EC_RECORD
        { sizeof(struct Request_ChunkSuspendWrite), 			&op_chunk_fake},			// CHUNK_CHECK_SUSPEND
        { sizeof(struct Request_ChunkFlushJournal), 			&op_chunk_fake},			// CHUNK_FLUSH_JOURNAL
        { sizeof(struct Request_ChunkGetBigJournalMeta), 		&op_chunk_fake}, 	// CHUNK_GET_BIG_JOURNAL_META
        { sizeof(Request_ChunkWriteECData2), 					&op_chunk_fake}, 			// CHUNK_WRITE_EC_DATA2
        { sizeof(struct Request_ChunkWriteDataToParity), 		&op_chunk_fake}, 	// CHUNK_WRITE_DATA_TO_PARITY
        { sizeof(struct Request_ChunkFlushECJournal), 			&op_chunk_fake}, 				// CHUNK_FLUSH_EC_JOURNAL
        { sizeof(struct Request_ChunkGetEcJournal), 			&op_chunk_fake}, 				// CHUNK_GET_EC_JOURNAL_META
        { sizeof(struct Request_ChunkWriteECGroupData), 		&op_chunk_fake},		// CHUNK_WRITE_ECGROUP_DATA2
        { sizeof(struct Request_ChunkWriteEC), 					&op_chunk_fake},				// CHUNK_WRITE_EC2
        { sizeof(struct Request_ChunkWriteECParity), 			&op_chunk_fake},			// CHUNK_WRITE_EC_PARITY_CHECK
        { sizeof(struct Request_ChunkCloseFd), 					&op_chunk_fake},				// CLOSE_CHUNK_FD
};

static void print_usage(char* name)
{
    printf("%s: 9 arguments are expected.\n", name);
}



// SIGTERM
static void terminate_gracefully(int sig)
{
    server_loop_terminate(g_chunk_ns);
}

bool restart_chunkserver = false;

static void sig_restart_handler(int signum)
{
    restart_chunkserver = true;
    server_loop_terminate(g_chunk_ns);
    signal(signum, sig_restart_handler);
}



// SIGRTMIN
extern int log_fd;
static void write_back_big_journal(int sig)
{
    extern bool write_back_by_hand;
    static char msg[] = "write_back_big_journal called\n";

    write(log_fd, msg, sizeof(msg));
    write_back_by_hand = true;
    signal(sig, write_back_big_journal);
}

// SIGRTMIN+1
static void set_current_journal_broken(int sig)
{
    extern bool current_journal_broken;
    static char msg[] = "set_current_journal_broken called\n";
    write(log_fd, msg, sizeof(msg));
    current_journal_broken = true;
    signal(sig, set_current_journal_broken);
}

// SIGRTMIN+2
static void set_stop_use_journal(int sig)
{
    extern bool stop_use_journal;
    static char msg[] = "set_stop_use_journal called\n";
    write(log_fd, msg, sizeof(msg));
    stop_use_journal = true;
    signal(sig, set_stop_use_journal);
}

extern bool update_full;
// SIGRTMIN+3
static void set_push_chunks_to_redis(int sig)
{
    update_full = true;
    signal(sig, set_push_chunks_to_redis);
}

// SIGRTMIN+4
static void change_report_disk_type(int sig)
{
    g_disk_type++;
    if(g_disk_type >= URSAX_DISK_TYPE_MAX){
        g_disk_type = 0;
    }
    signal(sig, change_report_disk_type);
}

//static int start_child(void)
//{
//    return -1;
//}

//extern int choose_index;
int process_io_args(char *str)
{
    LOG_INFO("process_io_args : %s ", str);
    extern int io_extra_flag;
    if(0 == strcasecmp(str, "o_sync")) {
        io_extra_flag |= O_SYNC;
    }
    else if(0 == strcasecmp(str, "o_direct")) {
        io_extra_flag |= O_DIRECT;
    }
    else if(0 == strcmp(str, "libaio")) {
        enable_libaio();
        io_extra_flag |= O_DIRECT;
    }
    else if(0 == strncmp(str, "ignore", sizeof("ignore") - 1)) {
        extern int ignore_sleep_utime;
        enable_ignore_ops();
        char *psize = str + sizeof("ignore");
        if(*(psize-1)) {
            ignore_sleep_utime = get_uint64(psize);
        }
    }
    else if(0 == strcmp(str, "none")) {
        // nothing
    }
    else if(0 == strcmp(str, "append")) {
        append_mode = true;
    }
    else if(0 == strcmp(str, "reporthdd")) {
        report_hdd = true;
    }
    else if(0 == strcmp(str, "4k")) {
        o_direct_size = 4096;
        LOG_INFO("set O_DIRECT_SIZE 4k");
    }
    else if(0 == strncmp(str, "maxsize", sizeof("maxsize")-1)){
        uint64_t size = parse_size(str + sizeof("maxsize"));
        set_journal_size(size, 0);
        // set_journal_rotate_size(size);
        LOG_INFO("GET journal max size %ld", size);
    }
    else if(0 == strncmp(str, "disktype", sizeof("disktype")-1)) {
        char *type = str + sizeof("disktype");
        if(0 == strcmp(type, "ssd")) {
            g_disk_type = URSAX_DISK_TYPE_SSD;
        }else if(0 == strcmp(type, "hdd")) {
            g_disk_type = URSAX_DISK_TYPE_HDD;
        }
    }
    else {
        LOG_ERROR("unknown io flag %s", str);
        exit(-1);
    }

    return 0;
}

int parse_io_args(const char *_arg)
{
    LOG_INFO("get io arg str: %s", _arg);
    char *arg = strdup(_arg);
    char *saveptr = NULL;
    char *sub_str = strtok_r(arg, ",", &saveptr);
    do {
        process_io_args(sub_str);
        sub_str = strtok_r(NULL, ",", &saveptr);
    } while(sub_str);
    free(arg);
    return 0;
}

static char** g_argv;

int main(int argc, char** argv)
{
    printf("Hello, UrsaX!\n");

    g_argv = argv;

    st_set_eventsys(ST_EVENTSYS_ALT);
    st_init();
    st_randomize_stacks(true);

    if (unlikely(argc != 10)){
        print_usage(argv[0]);
        return 0;
    }

    extern int debug_duration_key;
    st_key_create(&debug_duration_key, free);
    extern int debug_numbers_key;
    st_key_create(&debug_numbers_key, free);

#ifdef MAX_FD  // Don't set the max fd too large unless in debugging
    printf("aaa\n");
    set_max_open_fd(100000);
#endif

    int i = 1;  // parse cmdline arguments

    struct ChunkServerInfo csi;
    memset(&csi, 0, sizeof(csi));
    strncpy(csi.ip, argv[i++], sizeof(csi.ip));
    csi.port = strtol(argv[i++], NULL, 0);
    strncpy(csi.root, argv[i++], sizeof(csi.root));

    log_load_conf(argv[i]);
    char* log_conf = argv[i++];

    log_install_signal_handler(0);
    signal(SIGPIPE, SIG_IGN);
    signal(SIGTERM, terminate_gracefully); // exit sig
    signal(SIGRTMIN, write_back_big_journal);
    signal(SIGRTMIN+1, set_current_journal_broken);
    signal(SIGRTMIN+2, set_stop_use_journal);
    signal(SIGUSR1, sig_restart_handler); // exit sig
    signal(SIGRTMIN+3, set_push_chunks_to_redis);
    signal(SIGRTMIN+4, change_report_disk_type);

    log_set_output(argv[i]);
    char* log_output = argv[i++];

    parse_redis_servers(argv[i++]);  // redis连接字符串，如果含有多个ip:port，表示redis集群

    csi.interval = strtol(argv[i++], NULL, 10);

    parse_managers(argv[i++]);
    parse_io_args(argv[i++]);

    if (0 != gethostname(csi.hostname, HOST_NAME_MAX)) {
        /** if fail, use ip instead. */
        snprintf(csi.hostname, HOST_NAME_MAX, "%s", csi.ip);
    }
    csi.hostname[HOST_NAME_MAX] = '\0';

    LOG_INFO("service IP: %s", csi.ip);
    LOG_INFO("service port: %u", csi.port);
    LOG_INFO("root path: %s", csi.root);
    LOG_INFO("log config file: %s", log_conf);
    LOG_INFO("log output file: %s", log_output);
    LOG_INFO("redis server: %s", redis_ip);
    LOG_INFO("interval: %u", csi.interval);
    LOG_INFO("manager ip: %s", str_ip(manager_ip));
    LOG_INFO("manager port: %u", manager_port);
    LOG_INFO("hostname: %s", csi.hostname);

    int ret;
    ret = be_open_server_root(csi.root);
    if (unlikely(ret < 0)) {
        LOG_ERROR("error opening server root");
        return -1;
    }

    bool is_ssd = be_check_root_ssd_file();
    if(is_ssd){
        g_disk_type |= URSAX_DISK_TYPE_SSD;
    }

    ret = redis_client_init(true);
    if (unlikely(ret < 0)) {
        LOG_ERROR("error initializing redis client");
        return -1;
    }

    ret = check_manager_in_redis();
    LOG_INFO("check_manager_in_redis() returns %d", ret);

    if(append_mode || check_journal(csi.root)) {
        if(!report_hdd){
            g_disk_type |= URSAX_DISK_TYPE_JOURNAL;
        }
        struct ChunkServerJournal *journal =
                init_chunkserver_journal(csi.root, write_back_journal_cb);
        if(journal == NULL) {
            return -1;
        }
        set_big_journal(journal);
    }


    ret = heartbeat_init(&csi);
    if (unlikely(ret < 0)) {
        LOG_ERROR("error initializing heartbeat");
        return -1;
    }

    incubator_init();
    cconnection_pool_init(1000);
    chunk_operations_init();

    LOG_INFO("chunkserver starting...");
    g_chunk_ns = new_net_server_std("chunkserver", operation_table, LEN(operation_table), NULL);

    ret = ursax_statis_init();
    if (unlikely(ret < 0)) {
        LOG_ERROR("error initializing statistics");
    }

    init_debug_thread();

    g_chunk_ns->bindaddr = csi.ip;
    g_chunk_ns->port = csi.port;
    g_chunk_ns->handler = NULL; // default handler
    g_chunk_ns->wait_on_termination = false;

    LOG_INFO("netserver (g_chunk_ns) start");
    while(true){
        netserver(g_chunk_ns);
        if(!restart_chunkserver){
            break;
        }
//        int rstart_child = start_child();
//        if(rstart_child == 0){
//            break;
//        }
    }

    extern bool no_flush_journal;
    if(restart_chunkserver){
        no_flush_journal = true;
    }
    LOG_INFO("netserver(g_chunk_ns) end");

    g_chunk_ns->stop_all_service = true;
    int count = 0;
    while(g_chunk_ns->num_conns){
        LOG_INFO("wait all services close g_chunk_ns->num_conns %d",
                g_chunk_ns->num_conns);
        show_dead_thread();
        st_sleep(1);
        if(count++ > 20){
            break;
        }
    }

    int tmp = 0;
    while (1) {
        if (tmp++ > 1000000)
            break;

        st_sleep(0);
    }

    fini_debug_thread();

    chunk_operations_fini();
    LOG_INFO("chunk_operations_fini run over");
    cconnection_pool_fini();
    LOG_INFO("cconnection_pool_fini run over");
    incubator_fini();
    LOG_INFO("incubator_fini run over");
    ursax_statis_fini();
    LOG_INFO("ursax_statis_fini run over");
    heartbeat_fini(restart_chunkserver);
    LOG_INFO("heartbeat_fini run over");
    fini_chunkserver_journal(get_big_journal(USE_JOURNAL_DEFAULT));
    LOG_INFO("fini_chunkserver_journal run over");
    be_close_server_root();
    LOG_INFO("be_close_server_root run over");
    redis_client_fini();
    LOG_INFO("redis_client_fini run over");
    free(g_chunk_ns);



    return 0;
}